// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#pragma once

#include <iomanip>
#include <sstream>
#include <string>

#include <melon/dynamic.h>

#include <pollux/common/base/checked_arithmetic.h>
#include <pollux/common/strings/string_view.h>

namespace kumo::pollux {
    namespace tz {
        class TimeZone;
    }

    enum class TimestampPrecision : int8_t {
        kMilliseconds = 3, // 10^3 milliseconds are equal to one second.
        kMicroseconds = 6, // 10^6 microseconds are equal to one second.
        kNanoseconds = 9, // 10^9 nanoseconds are equal to one second.
    };

    struct TimestampToStringOptions {
        using Precision = TimestampPrecision;

        Precision precision = Precision::kNanoseconds;

        // Whether to add a leading '+' when year is greater than 9999.
        bool leadingPositiveSign = false;

        /// Whether to skip trailing zeros of fractional part. E.g. when true,
        /// '2000-01-01 12:21:56.129000' becomes '2000-01-01 12:21:56.129'.
        bool skipTrailingZeros = false;

        /// Whether padding zeros are added when the digits of year is less than 4.
        /// E.g. when true, '1-01-01 05:17:32.000' becomes '0001-01-01 05:17:32.000',
        /// '-03-24 13:20:00.000' becomes '0000-03-24 13:20:00.000', and '-1-11-29
        /// 19:33:20.000' becomes '-0001-11-29 19:33:20.000'.
        bool zeroPaddingYear = false;

        // The separator of date and time.
        char dateTimeSeparator = 'T';

        enum class Mode : int8_t {
            /// ISO 8601 timestamp format: %Y-%m-%dT%H:%M:%S.nnnnnnnnn for nanoseconds
            /// precision; %Y-%m-%dT%H:%M:%S.nnn for milliseconds precision.
            kFull,
            /// ISO 8601 date format: %Y-%m-%d.
            kDateOnly,
            /// ISO 8601 time format: %H:%M:%S.nnnnnnnnn for nanoseconds precision,
            /// or %H:%M:%S.nnn for milliseconds precision.
            kTimeOnly,
        };

        Mode mode = Mode::kFull;

        const tz::TimeZone *timeZone = nullptr;
    };

    /// Returns the max length of a converted string from timestamp.
    std::string::size_type getMaxStringLength(
        const TimestampToStringOptions &options);

    struct Timestamp {
    public:
        static constexpr int64_t kMillisecondsInSecond = 1'000;
        static constexpr int64_t kMicrosecondsInMillisecond = 1'000;
        static constexpr int64_t kMicrosecondsInSecond =
                kMicrosecondsInMillisecond * kMillisecondsInSecond;
        static constexpr int64_t kNanosecondsInMicrosecond = 1'000;
        static constexpr int64_t kNanosecondsInMillisecond = 1'000'000;
        static constexpr int64_t kNanosInSecond =
                kNanosecondsInMillisecond * kMillisecondsInSecond;
        // The number of days between the Julian epoch and the Unix epoch.
        static constexpr int64_t kJulianToUnixEpochDays = 2440588LL;
        static constexpr int64_t kSecondsInDay = 86400LL;

        // Limit the range of seconds to avoid some problems. Seconds should be
        // in the range [INT64_MIN/1000 - 1, INT64_MAX/1000].
        // Presto's Timestamp is stored in one 64-bit signed integer for
        // milliseconds, this range ensures that Timestamp's range in Pollux will not
        // be smaller than Presto, and can make Timestamp::toString work correctly.
        static constexpr int64_t kMaxSeconds =
                std::numeric_limits<int64_t>::max() / kMillisecondsInSecond;
        static constexpr int64_t kMinSeconds =
                std::numeric_limits<int64_t>::min() / kMillisecondsInSecond - 1;

        // Nanoseconds should be less than 1 second.
        static constexpr uint64_t kMaxNanos = 999'999'999;

        constexpr Timestamp() : seconds_(0), nanos_(0) {
        }

        Timestamp(int64_t seconds, uint64_t nanos)
            : seconds_(seconds), nanos_(nanos) {
            POLLUX_USER_DCHECK_GE(
                seconds, kMinSeconds, "Timestamp seconds out of range");
            POLLUX_USER_DCHECK_LE(
                seconds, kMaxSeconds, "Timestamp seconds out of range");
            POLLUX_USER_DCHECK_LE(nanos, kMaxNanos, "Timestamp nanos out of range");
        }

        /// Creates a timestamp from the number of days since the Julian epoch
        /// and the number of nanoseconds.
        static Timestamp fromDaysAndNanos(int32_t days, int64_t nanos);

        // date is the number of days since unix epoch.
        static Timestamp fromDate(int32_t date);

        // Returns the current unix timestamp (ms precision).
        static Timestamp now();

        static Timestamp create(const melon::Dynamic &obj) {
            auto seconds = obj["seconds"].asInt();
            auto nanos = obj["nanos"].asInt();
            return Timestamp(seconds, nanos);
        }

        int64_t getSeconds() const {
            return seconds_;
        }

        uint64_t getNanos() const {
            return nanos_;
        }

        // Keep it in header for getting inlined.
        int64_t toNanos() const {
            // int64 can store around 292 years in nanos ~ till 2262-04-12.
            // When an integer overflow occurs in the calculation,
            // an exception will be thrown.
            try {
                return checkedPlus(
                    checkedMultiply(seconds_, (int64_t) 1'000'000'000), (int64_t) nanos_);
            } catch (const std::exception &e) {
                POLLUX_USER_FAIL(
                    "Could not convert Timestamp({}, {}) to nanoseconds, {}",
                    seconds_,
                    nanos_,
                    e.what());
            }
        }

        // Keep it in header for getting inlined.
        int64_t toMillis() const {
            // We use int128_t to make sure the computation does not overflow since
            // there are cases such that seconds*1000 does not fit in int64_t,
            // but seconds*1000 + nanos does, an example is Timestamp::minMillis().

            // If the final result does not fit in int64_t we throw.
            __int128_t result =
                    (__int128_t) seconds_ * 1'000 + (int64_t) (nanos_ / 1'000'000);
            if (result < INT64_MIN || result > INT64_MAX) {
                POLLUX_USER_FAIL(
                    "Could not convert Timestamp({}, {}) to milliseconds",
                    seconds_,
                    nanos_);
            }
            return result;
        }

        // Keep it in header for getting inlined.
        int64_t toMillisAllowOverflow() const {
            // Similar to the above toMillis() except that overflowed integer is allowed
            // as result.
            auto result = seconds_ * 1'000 + (int64_t) (nanos_ / 1'000'000);
            return result;
        }

        // Keep it in header for getting inlined.
        int64_t toMicros() const {
            // We use int128_t to make sure the computation does not overflows since
            // there are cases such that a negative seconds*1000000 does not fit in
            // int64_t, but seconds*1000000 + nanos does. An example is
            // Timestamp(-9223372036855, 224'192'000).

            // If the final result does not fit in int64_t we throw.
            __int128_t result = static_cast<__int128_t>(seconds_) * 1'000'000 +
                                static_cast<int64_t>(nanos_ / 1'000);
            if (result < INT64_MIN || result > INT64_MAX) {
                POLLUX_USER_FAIL(
                    "Could not convert Timestamp({}, {}) to microseconds",
                    seconds_,
                    nanos_);
            }
            return result;
        }

        Timestamp toPrecision(const TimestampPrecision &precision) const {
            uint64_t nanos = nanos_;
            switch (precision) {
                case TimestampPrecision::kMilliseconds:
                    nanos = nanos / 1'000'000 * 1'000'000;
                    break;
                case TimestampPrecision::kMicroseconds:
                    nanos = nanos / 1'000 * 1'000;
                    break;
                case TimestampPrecision::kNanoseconds:
                    break;
            }
            return Timestamp(seconds_, nanos);
        }

        /// Exports the current timestamp as a std::chrono::time_point of millisecond
        /// precision. Note that the conversion may overflow since the internal
        /// `seconds_` value will need to be multiplied by 1000.
        ///
        /// If `allowOverflow` is true, integer overflow is allowed in converting
        /// to milliseconds.
        ///
        /// Due to the limit of pollux/external/date, throws if timestamp is outside of
        /// [-32767-01-01, 32767-12-31] range.
        std::chrono::time_point<std::chrono::system_clock, std::chrono::milliseconds>
        toTimePointMs(bool allowOverflow = false) const;

        static Timestamp fromMillis(int64_t millis) {
            if (millis >= 0 || millis % 1'000 == 0) {
                return Timestamp(millis / 1'000, (millis % 1'000) * 1'000'000);
            }
            auto second = millis / 1'000 - 1;
            auto nano = ((millis - second * 1'000) % 1'000) * 1'000'000;
            return Timestamp(second, nano);
        }

        static Timestamp fromMillisNoError(int64_t millis)
#if defined(__has_feature)
#if __has_feature(__address_sanitizer__)
      __attribute__((__no_sanitize__("signed-integer-overflow")))
#endif
#endif
        {
            if (millis >= 0 || millis % 1'000 == 0) {
                return Timestamp(millis / 1'000, (millis % 1'000) * 1'000'000);
            }
            auto second = millis / 1'000 - 1;
            auto nano = ((millis - second * 1'000) % 1'000) * 1'000'000;
            return Timestamp(second, nano);
        }

        static Timestamp fromMicros(int64_t micros) {
            if (micros >= 0 || micros % 1'000'000 == 0) {
                return Timestamp(micros / 1'000'000, (micros % 1'000'000) * 1'000);
            }
            auto second = micros / 1'000'000 - 1;
            auto nano = ((micros - second * 1'000'000) % 1'000'000) * 1'000;
            return Timestamp(second, nano);
        }

        static Timestamp fromMicrosNoError(int64_t micros)
#if defined(__has_feature)
#if __has_feature(__address_sanitizer__)
      __attribute__((__no_sanitize__("signed-integer-overflow")))
#endif
#endif
        {
            if (micros >= 0 || micros % 1'000'000 == 0) {
                return Timestamp(micros / 1'000'000, (micros % 1'000'000) * 1'000);
            }
            auto second = micros / 1'000'000 - 1;
            auto nano = ((micros - second * 1'000'000) % 1'000'000) * 1'000;
            return Timestamp(second, nano);
        }

        static Timestamp fromNanos(int64_t nanos) {
            if (nanos >= 0 || nanos % 1'000'000'000 == 0) {
                return Timestamp(nanos / 1'000'000'000, nanos % 1'000'000'000);
            }
            auto second = nanos / 1'000'000'000 - 1;
            auto nano = (nanos - second * 1'000'000'000) % 1'000'000'000;
            return Timestamp(second, nano);
        }

        static const Timestamp minMillis() {
            // The minimum Timestamp that toMillis() method will not overflow.
            // Used to calculate the minimum value of the Presto timestamp.
            constexpr int64_t kMin = std::numeric_limits<int64_t>::min();
            return Timestamp(
                kMinSeconds,
                (kMin % kMillisecondsInSecond + kMillisecondsInSecond) *
                kNanosecondsInMillisecond);
        }

        static const Timestamp maxMillis() {
            // The maximum Timestamp that toMillis() method will not overflow.
            // Used to calculate the maximum value of the Presto timestamp.
            constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
            return Timestamp(
                kMaxSeconds, kMax % kMillisecondsInSecond * kNanosecondsInMillisecond);
        }

        static const Timestamp min() {
            return Timestamp(kMinSeconds, 0);
        }

        static const Timestamp max() {
            return Timestamp(kMaxSeconds, kMaxNanos);
        }

        /// Our own version of gmtime_r to avoid expensive calls to __tz_convert.
        /// This might not be very significant in micro benchmark, but is causing
        /// significant context switching cost in real world queries with higher
        /// concurrency (71% of time is on __tz_convert for some queries).
        ///
        /// Return whether the epoch second can be converted to a valid std::tm.
        static bool epochToCalendarUtc(int64_t seconds, std::tm &out);

        /// Our own version of timegm to avoid expensive calls to __tz_convert.
        ///
        /// This function is guaranteed to give same result as std::timegm when it is
        /// successful.
        static int64_t calendarUtcToEpoch(const std::tm &tm);

        /// Truncates a Timestamp value to the specified precision.
        static Timestamp truncate(Timestamp ts, TimestampPrecision precision) {
            switch (precision) {
                case TimestampPrecision::kMilliseconds:
                    return Timestamp::fromMillis(ts.toMillis());
                case TimestampPrecision::kMicroseconds:
                    return Timestamp::fromMicros(ts.toMicros());
                case TimestampPrecision::kNanoseconds:
                    return ts;
                default:
                    POLLUX_UNREACHABLE();
            }
        }

        /// Converts a std::tm to a time/date/timestamp string in ISO 8601 format
        /// according to TimestampToStringOptions.
        /// @param startPosition the start position of pre-allocated memory to write
        /// string to.
        static StringView tmToStringView(
            const std::tm &tmValue,
            uint64_t nanos,
            const TimestampToStringOptions &options,
            char *const startPosition);

        /// Converts a timestamp to a time/date/timestamp string in ISO 8601 format
        /// according to TimestampToStringOptions.
        /// @param startPosition the start position of pre-allocated memory to write
        /// string to.
        static StringView tsToStringView(
            const Timestamp &ts,
            const TimestampToStringOptions &options,
            char *const startPosition);

        // Assuming the timestamp represents a time at zone, converts it to the GMT
        // time at the same moment. For example:
        //
        //  Timestamp ts{0, 0};
        //  ts.Timezone("America/Los_Angeles");
        //  ts.toString(); // returns January 1, 1970 08:00:00
        void toGMT(const tz::TimeZone &zone);

        /// Assuming the timestamp represents a GMT time, converts it to the time at
        /// the same moment at zone. For example:
        ///
        ///  Timestamp ts{0, 0};
        ///  ts.Timezone("America/Los_Angeles");
        ///  ts.toString(); // returns December 31, 1969 16:00:00
        void toTimezone(const tz::TimeZone &zone);

        /// A default time zone that is same across the process.
        static const tz::TimeZone &defaultTimezone();

        bool operator==(const Timestamp &b) const {
            return seconds_ == b.seconds_ && nanos_ == b.nanos_;
        }

        bool operator!=(const Timestamp &b) const {
            return seconds_ != b.seconds_ || nanos_ != b.nanos_;
        }

        bool operator<(const Timestamp &b) const {
            return seconds_ < b.seconds_ ||
                   (seconds_ == b.seconds_ && nanos_ < b.nanos_);
        }

        bool operator<=(const Timestamp &b) const {
            return seconds_ < b.seconds_ ||
                   (seconds_ == b.seconds_ && nanos_ <= b.nanos_);
        }

        bool operator>(const Timestamp &b) const {
            return seconds_ > b.seconds_ ||
                   (seconds_ == b.seconds_ && nanos_ > b.nanos_);
        }

        bool operator>=(const Timestamp &b) const {
            return seconds_ > b.seconds_ ||
                   (seconds_ == b.seconds_ && nanos_ >= b.nanos_);
        }

        void operator++() {
            if (nanos_ < kMaxNanos) {
                nanos_++;
                return;
            }
            if (seconds_ < kMaxSeconds) {
                seconds_++;
                nanos_ = 0;
                return;
            }
            POLLUX_USER_FAIL("Timestamp nanos out of range");
        }

        void operator--() {
            if (nanos_ > 0) {
                nanos_--;
                return;
            }
            if (seconds_ > kMinSeconds) {
                seconds_--;
                nanos_ = kMaxNanos;
                return;
            }
            POLLUX_USER_FAIL("Timestamp nanos out of range");
        }

        // Needed for serialization of FlatVector<Timestamp>
        operator StringView() const {
            return StringView("TODO: Implement");
        }

        std::string toString(const TimestampToStringOptions &options = {}) const {
            std::tm tm;
            POLLUX_USER_CHECK(
                epochToCalendarUtc(seconds_, tm),
                "Can't convert seconds to time: {}",
                seconds_);
            std::string result;
            result.resize(getMaxStringLength(options));
            const auto view = tmToStringView(tm, nanos_, options, result.data());
            result.resize(view.size());
            return result;
        }

        operator std::string() const {
            return toString();
        }

        operator melon::Dynamic() const {
            return melon::Dynamic(seconds_);
        }

        melon::Dynamic serialize() const {
            melon::Dynamic obj = melon::Dynamic::object;
            obj["seconds"] = seconds_;
            obj["nanos"] = nanos_;
            return obj;
        }

        // Pretty printer for gtest.
        friend void PrintTo(const Timestamp &timestamp, std::ostream *os) {
            *os << "sec: " << timestamp.seconds_ << ", ns: " << timestamp.nanos_;
        }

    private:
        int64_t seconds_;
        uint64_t nanos_;
    };

    void parseTo(melon::StringPiece in, ::kumo::pollux::Timestamp &out);

    template<typename T>
    void toAppend(const ::kumo::pollux::Timestamp &value, T *result) {
        result->append(value.toString());
    }
} // namespace kumo::pollux

namespace std {
    template<>
    struct hash<::kumo::pollux::Timestamp> {
        size_t operator()(const ::kumo::pollux::Timestamp value) const {
            return kumo::pollux::bits::hashMix(value.getSeconds(), value.getNanos());
        }
    };

    std::string to_string(const ::kumo::pollux::Timestamp &ts);

    template<>
    class numeric_limits<kumo::pollux::Timestamp> {
    public:
        static kumo::pollux::Timestamp min() {
            return kumo::pollux::Timestamp::min();
        }

        static kumo::pollux::Timestamp max() {
            return kumo::pollux::Timestamp::max();
        }

        static kumo::pollux::Timestamp lowest() {
            return kumo::pollux::Timestamp::min();
        }
    };
} // namespace std

namespace melon {
    template<>
    struct hasher<::kumo::pollux::Timestamp> {
        size_t operator()(const ::kumo::pollux::Timestamp value) const {
            return kumo::pollux::bits::hashMix(value.getSeconds(), value.getNanos());
        }
    };
} // namespace melon

namespace fmt {
    template<>
    struct formatter<kumo::pollux::TimestampToStringOptions::Precision>
            : formatter<int> {
        auto format(
            kumo::pollux::TimestampToStringOptions::Precision s,
            format_context &ctx) const {
            return formatter<int>::format(static_cast<int>(s), ctx);
        }
    };

    template<>
    struct formatter<kumo::pollux::TimestampToStringOptions::Mode>
            : formatter<int> {
        auto format(
            kumo::pollux::TimestampToStringOptions::Mode s,
            format_context &ctx) const {
            return formatter<int>::format(static_cast<int>(s), ctx);
        }
    };
} // namespace fmt
